package pers.mihao.quickstream.common.task.scheduler.isolated;

import lombok.extern.slf4j.Slf4j;
import pers.mihao.quickstream.common.task.dto.AutomaticTaskDTO;
import pers.mihao.quickstream.common.task.manager.AutomaticTaskManager;
import pers.mihao.quickstream.common.task.scheduler.*;
import pers.mihao.quickstream.common.util.CollectionUtil;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import static pers.mihao.quickstream.common.util.ThreadPoolNameUtil.getThreadFactory;

/**
 * 单机版任务调度（集群版参考quartz）
 * <a href="https://www.w3cschool.cn/quartz_doc/quartz_doc-1xbu2clr.html">...</a>
 *
 * @author mihao
 * @since 2024/11/18
 */
@Slf4j
public class AutomaticTaskScheduler implements SchedulerCenter, SchedulerTaskInvoker {

    /**
     * 定时取数据时间 （秒）
     */
    public static final long tobeSchedulerTime = 30L;

    @Resource
    AutomaticTaskManager automaticTaskManager;

    private final AtomicBoolean run = new AtomicBoolean(false);

    ScheduledThreadPoolExecutor readTobeSchedulerTaskExecutor = new ScheduledThreadPoolExecutor(1, r -> {
        Thread thread = new Thread(r);
        thread.setName("thread-submitTask");
        return thread;
    });

    ScheduledThreadPoolExecutor taskExecutorService = new ScheduledThreadPoolExecutor(12,
            getThreadFactory("TaskExecutorService"));

    public void start() {
        if (run.get()) {
            return;
        }
        if (run.compareAndSet(false, true)) {
            log.info("定时调度任务启动...");
            readTobeSchedulerTaskExecutor.scheduleAtFixedRate(this::submitTaskTobeScheduler,
                    0, tobeSchedulerTime * 1000 - 50, TimeUnit.MILLISECONDS);
        }
    }

    /**
     * 提交代执行的任务
     */
    private void submitTaskTobeScheduler() {
        List<AutomaticTaskDTO> taskDTOS = automaticTaskManager.pullTobeSchedulerTask(tobeSchedulerTime * 1000);
        log.info("获取待处理的任务... {}", taskDTOS.stream().map(AutomaticTaskDTO::getBizId).collect(Collectors.toList()));
        if (CollectionUtil.isEmpty(taskDTOS)) {
            return;
        }
        dispatcher(taskDTOS);
    }

    @Override
    public void onNewTaskCreate() {
        readTobeSchedulerTaskExecutor.schedule(this::submitTaskTobeScheduler, 0, TimeUnit.MILLISECONDS);
    }

    @Override
    public void dispatcher(List<AutomaticTaskDTO> taskDTOS) {
        if (taskExecutorService.getQueue().size() > 10) {
            // todo 待处理的任务告警
        }

        for (AutomaticTaskDTO taskDTO : taskDTOS) {
            CompletableFuture<TaskInvokeRes> future = doTask(taskDTO);
            future.whenComplete((success, throwable) -> {
                log.info("定时任务处理完毕 {} TaskInvoke res {}", taskDTO, success, throwable);
                automaticTaskManager.taskInvoke(taskDTO, success);
            });
        }
    }

    @Override
    public CompletableFuture<TaskInvokeRes> doTask(AutomaticTaskDTO taskDTO) {
        long now = System.currentTimeMillis();
        long nextInvokeTime = taskDTO.getNextInvokeTime().getTime();
        log.info("提交定时任务 {}", taskDTO.getBizId());
        CompletableFuture<TaskInvokeRes> future = new CompletableFuture<>();
        taskExecutorService.schedule(new AutomaticTaskAsyncRunner(taskDTO, future),
                Math.max(0, nextInvokeTime - now), TimeUnit.MILLISECONDS);
        return future;
    }


}
